package com.flink.es;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;

/**
 * @Author hujf
 * @Date 2022/8/29 10:17
 * @Description TODO
 */
public class HttpSinkFunction extends RichSinkFunction<RowData> {
    private String url;
    private String body;
    private String type;
    private SerializationSchema<RowData> serializer;
    private String[] fields;
    private DataType[] dataTypes;


    public HttpSinkFunction() {
    }

    public HttpSinkFunction(String url, String body, String type, SerializationSchema<RowData> serializer, String[] fields, DataType[] dataTypes) {
        this.url = url;
        this.body = body;
        this.type = type;
        this.serializer = serializer;
        this.fields = fields;
        this.dataTypes = dataTypes;
    }

    public static HttpSinkFunction.Builder builder() {
        return new HttpSinkFunction.Builder();
    }

    public static class Builder {
        private String url;
        private String body;
        private String type;
        private SerializationSchema<RowData> serializer;
        private String[] fields;
        private DataType[] dataTypes;

        public Builder () {

        }

        public Builder setUrl(String url) {
            this.url = url;
            return this;
        }

        public Builder setFields(String[] fields) {
            this.fields = fields;
            return this;
        }

        public Builder setBody(String body) {
            this.body = body;
            return this;
        }

        public Builder setType(String type) {
            this.type = type;
            return this;
        }

        public Builder setDataTypes(DataType[] dataTypes) {
            this.dataTypes = dataTypes;
            return this;
        }

        public Builder setDeserializer(SerializationSchema<RowData> serializer) {
            this.serializer = serializer;
            return this;
        }

        public HttpSinkFunction build() {
            if (StringUtils.isBlank(url) || StringUtils.isBlank(body) || StringUtils.isBlank(type)) {
                throw new IllegalArgumentException("params has null");
            }
            return new HttpSinkFunction(this.url, this.body, this.type, this.serializer,
                    this.fields, this.dataTypes);
        }

    }

}
